package com.xiaoniu.mc.client.producer;

import com.xiaoniu.mc.client.config.ProducerConfig;
import com.xiaoniu.mc.client.constant.SystemConstant;
import com.xiaoniu.mc.client.context.ApplicationContext;
import com.xiaoniu.mc.client.exception.XmqConfigException;
import com.xiaoniu.mc.client.message.SendMessageWrapper;
import com.xiaoniu.mc.util.PropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;

import java.io.UnsupportedEncodingException;

/**
 * @类描述：
 * @创建人：林继丰
 * @创建时间：2017/8/25 16:17
 * @版权：Copyright (c) 深圳市牛鼎丰科技有限公司-版权所有.
 */
public class XmqProducer implements Producer {

    private volatile DefaultMQProducer producer;

    public void initProducer(String instanceName) throws MQClientException {
        instanceName = StringUtils.isBlank(instanceName) ? SystemConstant.DEFAULT_PRODUCER_INSTANCE_NAME : instanceName;
        if (ApplicationContext.getInstanceNameMapProducer().get(instanceName) != null)
            return;
        String nameServer = String.valueOf(PropertiesUtil.getProperties().get("xmq.nameserver"));
        if (StringUtils.isBlank(nameServer)) {
            throw new XmqConfigException("nameServer没有找到", new Throwable());
        }
        producer = new DefaultMQProducer();
        producer.setNamesrvAddr(nameServer);

        String producerTopicQueueNum = String.valueOf(PropertiesUtil.getProperties().get("xmq.topic.queue.nums"));
        if (StringUtils.isNotBlank(producerTopicQueueNum)) {
            producer.setDefaultTopicQueueNums(Integer.valueOf(producerTopicQueueNum));
        } else {
            producer.setDefaultTopicQueueNums(SystemConstant.DEFAULT_TOPIC_QUEUE_NUMS);
        }

        String timeout = String.valueOf(PropertiesUtil.getProperties().get("xmq.producer.timeout"));
        if (StringUtils.isNotBlank(timeout)) {
            producer.setSendMsgTimeout(Integer.valueOf(timeout));
        } else {
            producer.setSendMsgTimeout(SystemConstant.DEFAULT_TIMEOUT);
        }

        producer.setInstanceName(instanceName);
        producer.start();
        ApplicationContext.getInstanceNameMapProducer().put(instanceName, this);
    }


    @Override
    public void send(ProducerConfig config, String message) throws UnsupportedEncodingException {
        SendMessageWrapper messageWrapper = getMessageWrapper(config, message);
        config.getProducerStrategy().sendMsgOrderly(messageWrapper, getProducer());
    }

    @Override
    public void sendOrderly(ProducerConfig config, String message) throws UnsupportedEncodingException {
        SendMessageWrapper messageWrapper = getMessageWrapper(config, message);
        config.getProducerStrategy().sendMsgOrderly(messageWrapper, getProducer());
    }

    private SendMessageWrapper getMessageWrapper(ProducerConfig config, String message) {
        SendMessageWrapper wrapper = new SendMessageWrapper(message, config.getTopic(), config.getTags(), config.getKeys(), config.getSendCallback());
        return wrapper;
    }

    public DefaultMQProducer getProducer() {
        return producer;
    }
    
}
